package com.flx.dis.registry.zk.service.impl;

import com.flx.dis.registry.zk.entity.ZkLock;
import com.flx.dis.registry.zk.server.ZkServer;
import com.flx.dis.registry.zk.service.ZkLockService;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkDataListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @Author Fenglixiong
 * @Create 2022/4/22 0:07
 * @Description
 **/
@Slf4j
@Service
public class ZkLockServiceImpl implements ZkLockService {

    private static final String rootPath = "zk-lock";

    @Autowired
    private ZkServer zkServer;

    public ZkLock lock(String lockId,long timeout)throws Exception{
        //首先创建锁的临时节点
        String lockPath = zkServer.createEphemeralSequential(rootPath+"/"+lockId,"w");
        ZkLock zkLock = new ZkLock(lockId,lockPath,false);
        //创建成功不一定就抢到锁了，需要进行激活判断
        zkLock = tryActiveLock(zkLock);
        if(!zkLock.isActive()){
            try{
                synchronized (zkLock){
                    zkLock.wait(timeout);
                }
            }catch (InterruptedException e){
                throw new RuntimeException(e);
            }
        }
        if(!zkLock.isActive()){
            throw new RuntimeException("lock timeout");
        }
        return zkLock;
    }

    /**
     * 尝试激活锁
     * @param lock
     * @return
     * @throws Exception
     */
    public ZkLock tryActiveLock(ZkLock lock)throws Exception{
        //获取所有的锁候选者
        List<String> lockList = zkServer.getChildren(rootPath)
                .stream()
                .sorted()
                .map(e->rootPath+"/"+e)
                .collect(Collectors.toList());
        String first = lockList.get(0);
        if(first.equals(lock.getPath())){
            lock.setActive(true);
        }else {
            String preNodePath = lockList.get(lockList.indexOf(lock.getPath())-1);
            zkServer.subscribeDataChange(preNodePath, new IZkDataListener() {
                @Override
                public void handleDataChange(String s, Object o) throws Exception {

                }

                @Override
                public void handleDataDeleted(String path) throws Exception {
                    log.info("节点被删除,path = {}",path);
                    ZkLock zkLock = tryActiveLock(lock);
                    synchronized (lock){
                        if(zkLock.isActive()){
                            lock.notify();
                        }
                    }
                    zkServer.unsubscribeDataChange(preNodePath,this);
                }
            });
        }
        return lock;
    }

    public void unlock(ZkLock zkLock)throws Exception{
        if(zkLock.isActive()){
            zkServer.delete(zkLock.getPath());
        }
    }

}
